package streamforce.workload;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Properties;
import java.util.Random;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.Stat;
import streamforce.crypto.message.PolicyMessage;
import streamforce.crypto.owner.EPOwnerClient;

/**
 * Generating policies of 5 different types:
 * 
 * T1: select * from StockEvent(stockId = x) - 1
 * T2: select * from StockEvent(stockId = x, y < timestamp < z) - 20
 * T3: select * from StockEvent(stockId = x, y < hour < z) - 11
 * T4: select * from StockEvent(stockId = x, timestamp % y = z) - 20
 * T5: select sum(price), sum(volume) from StockEvent(stockId=x).win:length_batch(y)  - 8 
 * T6: join price (with other conditions) - 40 (10 from T2, 10 from T4) 
 */
public class PolicyGeneratorAgg {
	static Random rand;
	static Log log = LogFactory.getLog(PolicyGeneratorAgg.class);
	static ArrayList<PolicyMessage> policyMessages = new ArrayList<PolicyMessage>(); 
	
	//<streamId> 
	public static void main(String[] args) throws Exception{
		int streamId = new Integer(args[0]).intValue();				
		rand = new Random(streamId); 
		
		Properties prop = new Properties();
		prop.load(new FileInputStream("attributes.properties"));					
		String[] ws = prop.getProperty("windowSizes").split(",");
		int aggType = new Integer(prop.getProperty("aggregateType")).intValue();
		int maxId = new Integer(prop.getProperty("maxUserId")).intValue();
		int[] windowSizes = new int[ws.length];
		for (int i=0; i<ws.length; i++){
			windowSizes[i] = new Integer(ws[i]).intValue();
		}
		
		String policyFile = prop.getProperty("policyFile"); 
		ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(policyFile+"_agg"+aggType+"_"+streamId));
		
		
		EPOwnerClient owner = new EPOwnerClient(streamId);			
		generateAgg(owner,streamId,0,windowSizes,aggType,maxId);
		oos.writeObject(policyMessages);
		oos.flush(); 
		oos.close(); 
	}
		
	
	//type 5 - 8
	public static void generateAgg(EPOwnerClient owner, int streamId,
			int userIdBase, int[] windowSizes, int aggType, int maxId) {
		PolicyInfo[] s = new PolicyInfo[windowSizes.length];
		PolicyMessage pm;
		int ws; 
		for (int i = 0; i < s.length; i++) {
			ws = windowSizes[i];
			s[i] = new PolicyInfo(streamId, userIdBase + i, "stockId_"
					+ streamId + " and ws_"
					+ ws, "stockId="+streamId, ws);
			long start = System.currentTimeMillis(); 	
			if (aggType==2)
				pm = EPOwnerClient.initAggregateClient_2(owner, streamId, s[i]);
			else if (aggType==1)
				pm = EPOwnerClient.initAggregateClient_1(owner, streamId, s[i]);
			else
				pm = EPOwnerClient.initAggregateClient_3(owner, streamId, s[i],maxId);
			policyMessages.add(pm); 
			long end = System.currentTimeMillis(); 			
			log.info("policy "+s[i].getUserId()+": "+s[i].getPolicy());
		}		
		
		
	}
	
	
}
